%% copy riak-0.14.2 source to learn dynamo implementation and erlang
%% author: lichuang
%% copy source from riak_core_ring.erl

%% @doc edynamo_ring manages a edynamo node's local view of partition ownership.
%%      The functions in this module revolve around use of the chstate record,
%%      which should be treated as opaque by other modules.  Riak nodes exchange
%%      instances of these records via gossip in order to converge on a common
%%      view of node/partition ownership.
-module(edynamo_ring).

-export([fresh/0, all_members/1, my_indices/1, preflist/2,
    all_owners/1, random_other_index/1, random_other_index/2,
    num_partitions/1, get_meta/2, update_meta/3,
    random_node/1, random_other_node/1, index_owner/2,
    transfer_node/3, all_preflists/2]).

% @type riak_core_ring(). The opaque data type used for partition ownership.
-record(chstate, {
    nodename, % the Node responsible for this chstate
    vclock,   % for this chstate object, entries are {Node, Ctr}
    chring,   % chash ring of {IndexAsInt, Node} mappings
    meta      % dict of cluster-wide other data (primarily bucket N-value, etc)
}).
%-type riak_core_ring() :: #chstate{}.
-type chstate() :: #chstate{}.
% @type meta_entry(). Record for each entry in #chstate.meta
-record(meta_entry, {
    value,    % The value stored under this entry
    lastmod   % The last modified time of this entry, 
	      %  from calendar:datetime_to_gregorian_seconds(
	      %                             calendar:universal_time()), 
  }).

% @doc This is used only when this node is creating a brand new cluster.
-spec fresh() -> chstate().
fresh() ->
  fresh(node()).

% @doc Equivalent to fresh/0 but allows specification of the local node name.
%      Called by fresh/0, and otherwise only intended for testing purposes.
-spec fresh(NodeName :: term()) -> chstate().
fresh(NodeName) ->
  fresh(app_helper:get_env(edynamo, ring_creation_size), NodeName).

% @doc Equivalent to fresh/1 but allows specification of the ring size.
%      Called by fresh/1, and otherwise only intended for testing purposes.
-spec fresh(RingSize :: integer(), NodeName :: term()) -> chstate().
fresh(RingSize, NodeName) ->
  #chstate{nodename = NodeName,
	   vclock = vclock:fresh(),
	   chring = chash:fresh(RingSize, NodeName),
	   meta = dict:new()}.

-spec all_members(State::chstate()) -> [Node :: term()].
all_members(State) ->
  chash:members(State#chstate.chring).

-spec my_indices(State::chstate()) -> [integer()].
my_indices(State) ->
  [I || {I, Owner} <- ?MODULE:all_owners(State), Owner =:= node()].

% @doc Provide all ownership information in the form of {Index,Node} pairs.
-spec all_owners(State :: chstate()) -> [{Index :: integer(), Node :: term()}].
all_owners(State) ->
  chash:nodes(State#chstate.chring).

% @doc Return a partition index not owned by the node executing this function.
%      If this node owns all partitions, return any index.
-spec random_other_index(State :: chstate()) -> integer().
random_other_index(State) ->
  L = [I || {I, Owner} <- ?MODULE:all_owners(State), Owner =/= node()],
  case L of
    [] -> hd(my_indices(State));
    _  -> lists:nth(random:uniform(length(L)), L)
  end.

-spec random_other_index(State :: chstate(), Exclude :: [term()]) -> integer() | no_indices.
random_other_index(State, Exclude) when is_list(Exclude) ->
  L = [I || {I, Owner} <- ?MODULE:all_owners(State),
	    Owner =/= node(),
	    not lists:member(I, Exclude)],
  case L of
    [] -> no_indices;
    _  -> lists:nth(random:uniform(length(L)), L)
  end.

% @doc Return the number of partitions in this Riak ring.
-spec num_partitions(State :: chstate()) -> integer().
num_partitions(State) ->
  chash:size(State#chstate.chring).

% @doc Return a value from the cluster metadata dict
-spec get_meta(Key :: term(), State :: chstate()) -> 
  {ok, term()} | undefined.
get_meta(Key, State) ->
  case dict:find(Key, State#chstate.meta) of
    error -> undefined;
    {ok, M} -> {ok, M#meta_entry.value}
  end.

% @doc For a given object key, produce the ordered list of
%      {partition,node} pairs that could be responsible for that object.
-spec preflist(Key :: binary(), State :: chstate()) ->
                               [{Index :: integer(), Node :: term()}].
preflist(Key, State) ->
  chash:successors(Key, State#chstate.chring).

% @doc Set a key in the cluster metadata dict
-spec update_meta(Key :: term(), Val :: term(), State :: chstate()) -> chstate().
update_meta(Key, Val, State) ->
  Change = case dict:find(Key, State#chstate.meta) of
            {ok, OldM} ->
              Val /= OldM#meta_entry.value;
            error ->
              true
           end,
  if Change ->           
      M = #meta_entry {
        lastmod = calendar:datetime_to_gregorian_seconds(
                    calendar:universal_time()),
        value = Val},
      VClock = vclock:increment(State#chstate.nodename,
                                State#chstate.vclock),
      State#chstate{vclock = VClock,
                    meta = dict:store(Key, M, State#chstate.meta)};
    true ->
      State
  end.      

% @doc Return a randomly-chosen node from amongst the owners.
-spec random_node(State :: chstate()) -> Node :: term().
random_node(State) ->
  L = all_members(State),
  lists:nth(random:uniform(length(L)), L).

% @doc Return a randomly-chosen node from amongst the owners other than this one.
-spec random_other_node(State :: chstate()) -> Node :: term() | no_node.
random_other_node(State) ->
  case lists:delete(node(), all_members(State)) of
    [] ->
      no_node;
    L ->
      lists:nth(random:uniform(length(L)), L)
  end.

% @doc Return the node that owns the given index.
-spec index_owner(State :: chstate(), Idx :: integer()) -> Node :: term().
index_owner(State, Idx) ->
  hd([Owner || {I, Owner} <- ?MODULE:all_owners(State), I =:= Idx]).

% @doc Provide every preflist in the ring, truncated at N.
-spec all_preflists(State :: chstate(), N :: integer()) ->
                               [[{Index :: integer(), Node :: term()}]].
all_preflists(State, N) ->
  [lists:sublist(preflist(Key, State), N) ||
    Key <- [<<(I + 1) : 160 / integer>> ||
              {I, _Owner} <- ?MODULE:all_owners(State)]].

-spec transfer_node(Idx :: integer(), Node :: term(), MyState :: chstate()) -> chstate().
transfer_node(Idx, Node, MyState) ->
  case chash:lookup(Idx, MyState#chstate.chring) of
    Node ->
      MyState;
    _ ->
      Me = MyState#chstate.nodename,
      VClock = vclock:increment(Me, MyState#chstate.vclock),
      CHRing = chash:update(Idx, Node, MyState#chstate.chring),
      #chstate{nodename=Me,vclock=VClock,chring=CHRing,
        meta=MyState#chstate.meta}
  end.
